package com.rds.benchmark.connector;

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;


public class SingleLettuceConnector implements JedisConnector {

    private GenericObjectPool<StatefulRedisConnection<String, String>> pool;
    private String stringDes;

    public SingleLettuceConnector(final String host, final int port, final int max_threads, final String pass) {
        RedisURI redisUri = null;
        if (pass == null)
            redisUri = RedisURI.builder()
                    .withHost(host)
                    .withPort(port)
                    .build();
        else
            redisUri = RedisURI.builder()
                    .withHost(host)
                    .withPort(port)
                    .withPassword(pass)
                    .build();
        RedisClient redisClient = RedisClient.create(redisUri);

        GenericObjectPoolConfig objectGenericObjectPoolConfig = new GenericObjectPoolConfig<>();
        objectGenericObjectPoolConfig.setMaxTotal(max_threads + 1);
        objectGenericObjectPoolConfig.setMaxIdle(max_threads + 1);

        pool = ConnectionPoolSupport
                .createGenericObjectPool(() -> redisClient.connect(), objectGenericObjectPoolConfig);


        stringDes = "SingleLettuceConnector(host='" + host + "', port='" + port + "' )";
        System.out.println(stringDes + " connecting....");

    }

    @Override
    public Object getJedis() {
        try {
            return pool.borrowObject();
        } catch (Exception e) {

        }
        return null;
    }

    @Override
    public boolean test() {
        try (StatefulRedisConnection<String, String> connection = pool.borrowObject()) {
            RedisCommands<String, String> commands = connection.sync();
            return "PONG".equalsIgnoreCase(commands.ping());
        } catch (Exception e) {

        }
        return false;
    }

    @Override
    public String type() {
        return "SINGLE";
    }

    @Override
    public void close(Object o){
        if(o instanceof RedisCommands){
            StatefulRedisConnection statefulConnection = ((RedisCommands) o).getStatefulConnection();
            pool.returnObject(statefulConnection);
        }
    }
}
